package com.hexiang.rocketmq.mq.producer.transaction.listener;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.hexiang.rocketmq.message.DemoMessage;
import com.hexiang.rocketmq.service.DemoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 〈事务消息的监听器〉<br>
 */
@Component
@RocketMQTransactionListener
@Slf4j
public class DemoSpringTransactionListener implements RocketMQLocalTransactionListener {

    @Resource
    private DemoService demoService;
    @Resource
    private ObjectMapper objectMapper;

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object arg) {

        RocketMQLocalTransactionState transactionState = RocketMQLocalTransactionState.COMMIT;
        //获取消息体
        String jsonString = new String((byte[]) message.getPayload());
        log.info("demoExecuteLocalTransactionStart,message is {}", jsonString);
        if (StringUtils.isNotBlank(jsonString)) {
            DemoMessage demoMessage;
            try {
                demoMessage = objectMapper.readValue(jsonString, DemoMessage.class);
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException("事务消息消息转对象失败");
            }
            //判断是否为空
            if (null == demoMessage) {
                throw new IllegalArgumentException("事务消息消息对象不能为空");
            }
            // 入库单
            try {
                //保存入库物料信息、保存入库图片信息，事务方法
                demoService.saveDemo(demoMessage);
            } catch (RuntimeException e) {
                log.error("executeLocalTransactionRuntimeException,message is{}", message, e);
                transactionState = RocketMQLocalTransactionState.ROLLBACK;
            } catch (Exception e) {
                log.error("executeLocalTransactionException,message is{}", message, e);
                transactionState = RocketMQLocalTransactionState.UNKNOWN;
            }
        } else {
            throw new IllegalArgumentException("事务消息消息对象不能为空");
        }

        log.info("demoExecuteLocalTransactionEnd,message is {},transactionState is {}", message, transactionState);

        return transactionState;

    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {

        String jsonString = new String((byte[]) message.getPayload());

        log.info("demoCheckLocalTransactionStart,message is {}", jsonString);

        RocketMQLocalTransactionState transactionState = RocketMQLocalTransactionState.ROLLBACK;

        if (StringUtils.isNotBlank(jsonString)) {
            DemoMessage demoMessage = null;
            //获取消息对象
            try {
                demoMessage = objectMapper.readValue(jsonString, DemoMessage.class);
            } catch (JsonProcessingException e) {
                throw new IllegalArgumentException("事务消息消息转对象失败");
            }

            Long id;
            if (null != demoMessage) {
                //入库单号
                id = demoMessage.getId();
                if (null != id) {
                    try {
                        //检查入库单信息是否存在
                        if (demoService.checkDemoMessageExist(demoMessage)) {
                            transactionState = RocketMQLocalTransactionState.COMMIT;
                        }
                    } catch (Exception e) {
                        log.error("checkLocalTransactionException", e);
                        transactionState = RocketMQLocalTransactionState.UNKNOWN;
                    }
                } else {
                    log.warn("checkLocalTransactionDemoMessageIsNull,messageObj is {}", demoMessage);
                }
            } else {
                log.warn("checkLocalTransactionDemoMessageIsNoValid,message is {}", jsonString);
            }
        } else {
            log.warn("checkLocalTransactionDemoMessageIsNull,message is {}", jsonString);
        }

        log.info("demoCheckLocalTransactionEnd,message is {},transactionState is {}", message, transactionState);

        return transactionState;
    }
}